package me.tuyou.stream;

import me.tuyou.common.MyLog;
import me.tuyou.common.MyLogGenerator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;

public class TriggerTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new MyLogGenerator())
                .keyBy(MyLog::getDeviceId)
                .timeWindow(Time.seconds(10))
                .trigger(new MyTrigger())
                .sum("count")
                .print();
        env.execute();
    }

    public static class MyTrigger extends Trigger {

        int count = 0;

        @Override
        public TriggerResult onElement(Object element, long timestamp, Window window, TriggerContext ctx) throws Exception {
            if (count > 9) {
                count = 0;
                System.out.println("触发器触发");
                return TriggerResult.FIRE;
            } else {
                count++;
                System.out.println("onElement: " + element + ", count:" + count);
                return TriggerResult.CONTINUE;
            }
        }

        @Override
        public TriggerResult onProcessingTime(long time, Window window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long time, Window window, TriggerContext ctx) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public void clear(Window window, TriggerContext ctx) throws Exception {

        }
    }
}
